package worker

import (
	"context"
	"gitee.com/zhengjianfeng/crontab/common"
	"github.com/mongodb/mongo-go-driver/mongo"
	"github.com/mongodb/mongo-go-driver/mongo/clientopt"
	"time"
)

// mongodb存储日志
type LogSink struct {
	client         *mongo.Client
	logCollection  *mongo.Collection
	logChan        chan *common.JobLog
	autoCommitChan chan *common.LogBatch
}

var (
	G_logSink *LogSink
)

// 批量写入日志
func (logSink *LogSink) saveLogs(batch *common.LogBatch) {
	_, _ = logSink.logCollection.InsertMany(context.TODO(), batch.Logs)
}

// 日志存储协程
func (logSink *LogSink) writeLook() {
	var (
		log          *common.JobLog
		logBatch     *common.LogBatch // 当前的批次
		commitTimer  *time.Timer
		timeOutBatch *common.LogBatch // 超时批次
	)

	for {
		select {
		case log = <-logSink.logChan:
			// 把这条记录写到mongodb中
			if logBatch == nil {
				logBatch = &common.LogBatch{}
				// 让这个批次超时自动提交(给1秒时间)
				commitTimer = time.AfterFunc(time.Duration(G_config.JobLogCommitTimeout)*time.Millisecond, func(logBatch *common.LogBatch) func() {
					// 发出超时通知，不要提交batch
					return func() {
						logSink.autoCommitChan <- logBatch
					}
				}(logBatch))
			}

			// 把新的日志追加到批次中
			logBatch.Logs = append(logBatch.Logs, log)
			// 如果批次满了，就立即发送
			if len(logBatch.Logs) >= G_config.JobLogBatchSize {
				// 发送日志
				logSink.saveLogs(logBatch)

				// 清空logBatch
				logBatch = nil

				// 取消定时器
				commitTimer.Stop()
			}
		case timeOutBatch = <-logSink.autoCommitChan: // 过期的批次
			// 过期批次是否仍就是当前批次
			if timeOutBatch != logBatch {
				continue // 跳过已经被提交的批次
			}

			// 把批次写入到mongodb中
			logSink.saveLogs(logBatch)

			// 清空logBatch
			logBatch = nil
		}
	}
}

func InitLogSink() (err error) {

	var (
		client *mongo.Client
	)

	// 建立mongodb连接
	if client, err = mongo.Connect(context.TODO(), G_config.MongodbUri, clientopt.ConnectTimeout(time.Duration(G_config.MongodbConnectTimeout)*time.Millisecond)); err != nil {
		return
	}

	// 选择db 和 collection
	G_logSink = &LogSink{
		client:         client,
		logCollection:  client.Database("cron").Collection("log"),
		logChan:        make(chan *common.JobLog, 1000),
		autoCommitChan: make(chan *common.LogBatch, 1000),
	}

	// 启动一个mongodb处理协程
	go G_logSink.writeLook()

	return
}

// 发送日志
func (logSink *LogSink) Append(jobLog *common.JobLog) {

	select {
	case logSink.logChan <- jobLog:

	default:
		// 队列满了就丢弃

	}

}
